package com.shujia.spark.streaming

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.dstream.DStream
import java.util.Date

import org.apache.spark.storage.StorageLevel

object Demo7DriverHA {
  def main(args: Array[String]): Unit = {


    val conf = new SparkConf().setAppName("wc")

    val sc = new SparkContext(conf)

    //创建sparkStraming上下文对象,  需要指定创建RDD的间隔时间    batch时间   可以理解为多少秒计算一次
    val ssc = new StreamingContext(sc, Durations.seconds(5))
    //hdfs路径
    //    ssc.checkpoint("/data/checkpoint")

    //连接实时数据源，创建DStream,
    //相当于socket的client
    //可以在linux中启动一个服务端    ，  nc -lk 8888
    val linesDS: DStream[String] = ssc.socketTextStream("node1", 8888)

    /**
      * Dstream的代码逻辑，每个5秒会被计算一次，每次计算都是独立的
      *
      */
    val countDS: DStream[(String, Int)] = linesDS
      .flatMap(_.split(","))
      .map((_, 1))
      .reduceByKey(_ + _)

    //相当于一个action算子
    countDS.print()


    //启动sparkstreaming程序
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }
}
